package flink.http_connect;

import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.concurrent.TimeUnit;

public class AsyncHttpConnector {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.fromElements("param1","param2","param3");

        SingleOutputStreamOperator<R> result = AsyncDataStream.unorderedWait(
                lines,                          //输入的数据流
                new AsyncHttpFunction(),   //异步查询的Function实例
                2000,                   //超时时间
                TimeUnit.MILLISECONDS,         //时间单位
                10                     //最大异步并发请求数量（并发的线程队列数）
        );

        result.print();

        env.execute("HttpAsyncQuery");
    }
}
